import { noop, OnDestory, OnInit } from '@core/types'
import { Connection, Channel, connect, ConsumeMessage } from 'amqplib'
import { v4 } from 'uuid';
export class RabbitmqServer implements OnDestory, OnInit {
    public mq: Connection | undefined;
    private channels: Channel[] = []
    private closer: Function[] = [];
    private intervalHandler: any;
    async onInit(): Promise<void> {
        const mqUrl = process.env.MQ_URL || ``
        this.mq = await connect(mqUrl)
        this.mq.on('error', e => {
            this.onInit()
        })
        console.info(`RabbitmqServerApplication inited`)
    }

    /**
     * 创建stream房间
     * @param room 房间名称
     */
    async createRoom(room: string, handler: (msg: ConsumeMessage, c: Channel) => void) {
        const channel = await this.createChannel()
        channel.on('error', e => this.createRoom(room, handler))
        await channel.assertExchange(room, 'fanout')
        // x-dead-letter-exchange
        await channel.assertQueue(room, {
            durable: true,
            autoDelete: false,
            exclusive: false,
            arguments: {
                'x-queue-type': 'stream'
            }
        })
        await channel.prefetch(100)
        await channel.bindQueue(room, room, '')
        await this.sendMsgToRoom(room, 'hello')
        const consume = await channel.consume(room, (msg) => {
            if (!msg) return;
            handler(msg, channel)
        }, {
            noAck: false,
            arguments: {
                // 'x-stream-offset': { "!": `timestamp`, value: dayjs().second() }
                'x-stream-offset': "1s"
            }
        })
        this.closer.push(() => channel.cancel(consume.consumerTag))
    }
    /**
     * 往房间里 发送消息
     */
    async sendMsgToRoom(room: string, msg: string | Buffer) {
        if (!this.mq) throw new Error(`mq is not connected`)
        const channel = await this.mq.createChannel()
        const msgId = v4()
        channel.publish(room, '', Buffer.from(msg), { persistent: true })
        await channel.close()
        return msgId;
    }
    /**
     * 定时任务
     */
    async createDelay(topic: string) {
        const channel = await this.createChannel()
    }

    async sendMsgToDelay(topic: string, runAt: Date, msg: string | Buffer) {

    }
    async createChannel() {
        if (!this.mq) throw new Error(`mq is not connected`)
        const channel = await this.mq.createChannel()
        channel.on('close', () => {
            console.log(`mq connection is closed`)
            this.channels = this.channels.filter(c => c !== channel)
        })
        this.channels.push(channel)
        return channel;
    }

    async onDestory(): Promise<void> {
        await Promise.all(this.closer.map(c => c()))
        await Promise.all(this.channels.map(async c => {
            const stream = Reflect.get(c.connection, `stream`);
            console.log(`stream connecting is ${stream.connecting}`)
            if (stream.connecting) {
                await c.close().catch(noop)
            }
        })).catch(noop)
        if (!this.mq) return;
        const stream = Reflect.get(this.mq.connection, `stream`);
        console.log(`stream connecting is ${stream.connecting}`)
        if (stream.connecting) {
            await this.mq.close().catch(noop)
        }
        if (this.intervalHandler) clearInterval(this.intervalHandler)
        console.info(`RabbitmqServerApplication destoried`)
    }
}